package com.study.skywalking.component;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.*;

/**
 * Created by zhangbo on 2020/6/10.
 */
@Slf4j
@Component
public class RocketProduct implements InitializingBean {

    @Value("${ROCKET.PRODUCE.GROUP:studySkywalking}")
    private String produceGroup;

    @Value("${ROCKET.NAME.SERVER}")
    private String nameServer;


    private volatile TransactionMQProducer product;

    private ExecutorService executorService;

    private volatile boolean isReady;

    public void init() {
        isReady = false;
        this.product = new TransactionMQProducer(produceGroup);
        this.product.setNamesrvAddr(nameServer);

        this.executorService = new ThreadPoolExecutor(5, 100, 10, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(1000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName(produceGroup + "- thread");
                thread.setDaemon(true);
                return thread;
            }
        });
        this.product.setExecutorService(executorService);
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        try {
            init();
            this.product.start();
            log.info("rocket启动完成");
            isReady = true;
        } catch (Exception e) {
            log.error("RocketMq启动异常：{}", e);
            isReady = false;
        }
    }

    public SendResult sendMessage(Message message) {
        if (!isReady) {
            throw new RuntimeException("RocketMq未完成启动！");
        }
        try {
            SendResult send = this.product.send(message);
            return send;
        } catch (MQClientException e) {
            log.error("mq Client异常：{}", e);
        } catch (RemotingException e) {
            log.error("mq Remoting异常：{}", e);
        } catch (MQBrokerException e) {
            log.error("mq Broker异常：{}", e);
        } catch (InterruptedException e) {
            log.error("mq异常，线程中断：{}", e);
            Thread.currentThread().interrupt();
        }
        return null;
    }

    /**
     * 将消息发送到指定队列
     * @param message
     * @param queueNumber
     * @return
     */
    public SendResult sendMessage(Message message,final int queueNumber) {
        if (!isReady) {
            throw new RuntimeException("RocketMq未完成启动！");
        }
        try {
            SendResult send = this.product.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    int count = queueNumber % list.size();
                    return list.get(count);
                }
            },queueNumber);
            return send;
        } catch (MQClientException e) {
            log.error("mq Client异常：{}", e);
        } catch (RemotingException e) {
            log.error("mq Remoting异常：{}", e);
        } catch (MQBrokerException e) {
            log.error("mq Broker异常：{}", e);
        } catch (InterruptedException e) {
            log.error("mq异常，线程中断：{}", e);
            Thread.currentThread().interrupt();
        }
        return null;
    }

    public void shutdown() {
        this.product.shutdown();
    }

}
